package org.sn.jdish.spark.operator;

import java.util.Arrays;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;

/**
 * 对元素进行过滤，对每个元素应用f函数，返回值为true的元素在RDD中保留，返回为false的将过滤掉
 * 
 * @author snzigod@hotmail.com
 *
 */
public class Filter {

	public static void main(String[] args) {
		/**
		 * SparkConf:第一步创建一个SparkConf，在这个对象里面可以设置允许模式Local Standalone yarn
		 * AppName(可以在Web UI中看到) 还可以设置Spark运行时的资源要求
		 */
		SparkConf conf = new SparkConf().setAppName("sparkOperator").setMaster("local");

		/**
		 * 基于SparkConf的对象可以创建出来一个SparkContext Spark上下文
		 * SparkContext是通往集群的唯一通道，SparkContext在创建的时候还会创建任务调度器
		 */
		JavaSparkContext sc = new JavaSparkContext(conf);

		filter(sc);

		sc.close();
	}

	static void filter(JavaSparkContext sc) {
		List<Integer> datas = Arrays.asList(1, 2, 3, 7, 4, 5, 8);

		JavaRDD<Integer> rddData = sc.parallelize(datas);
		rddData.filter(
				// jdk1.8
				// v1 -> v1 >= 3
				new Function<Integer, Boolean>() {
					private static final long serialVersionUID = 1L;

					public Boolean call(Integer v) throws Exception {
						// 去除>=4的数
						return v >= 4;
					}
				}).foreach(
						// jdk1.8
						// v -> System.out.println(v)
						new VoidFunction<Integer>() {
							private static final long serialVersionUID = 1L;

							@Override
							public void call(Integer integer) throws Exception {
								System.out.println(integer);
							}
						});
	}

}
